package com.atguigu.medical.realtime.app.dwd;

import com.atguigu.medical.realtime.app.BaseSQLApp;
import com.atguigu.medical.realtime.common.Constant;
import com.atguigu.medical.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/5/26 14:44
 */
public class DwdDiagnosisPrescription extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdDiagnosisPrescription().init(
            30001,
            2,
            "DwdDiagnosisPrescription"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 因为有 join, 所以需要设置 ttl
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        
        // 1. 读取 ods_medical 表
        readOdsMedical(tEnv, "DwdDiagnosisPrescription");
        // 2. 使用 jdbc 连接从 mysql 读取药品表:medicine
        tEnv.executeSql("create table medicine(" +
                            " id bigint, " +
                            " price decimal(16,2) " +
                            ")with(" +
                            " 'connector' = 'jdbc'," +
                            " 'url' = 'jdbc:mysql://hadoop162:3306/medical?useSSL=false'," +
                            " 'table-name' = 'medicine', " +
                            " 'lookup.cache' = 'PARTIAL'," +
                            " 'lookup.partial-cache.max-rows' = '1000'," +
                            " 'lookup.partial-cache.expire-after-access' = '180 minute', " +
                            " 'username' = 'root', " +
                            " 'password' = 'aaaaaa' " +
                            ")");
        // 3. 过滤出处方表
        Table prescriptionTable = tEnv
            .sqlQuery("select " +
                          " after['id'] id,  " +
                          " after['doctor_id'] doctor_id,  " +
                          " after['patient_id'] patient_id,  " +
                          " after['consultation_id'] consultation_id,  " +
                          " after['create_time'] prescription_time,  " +
                          " after['instruction'] instruction,  " +
                          " after['total_amount'] total_amount  " +
                          "from ods_medical " +
                          "where source['db']='medical' " +
                          "and source['table']='prescription' " +
                          "and op='c' ");
        tEnv.createTemporaryView("prescription", prescriptionTable);
        
        // 4. 过滤出处方明细表
        Table prescriptionDetailTable = tEnv
            .sqlQuery("select " +
                          " after['id'] id,  " +
                          " after['prescription_id'] prescription_id,  " +
                          " after['instruction'] instruction,  " +
                          " after['medicine_id'] medicine_id,  " +
                          " after['count'] ct," +
                          " pt," +
                          " ts  " +
                          "from ods_medical " +
                          "where source['db']='medical' " +
                          "and source['table']='prescription_detail' " +
                          "and op='c' ");
        tEnv.createTemporaryView("prescription_detail", prescriptionDetailTable);
        // 5. 3 张表进行 join
        Table result = tEnv
            .sqlQuery("select " +
                          " pre_detail.id  pre_detail_id, " +
                          " pre.id  prescription_id, " +
                          " pre_detail.instruction  pre_detail_instruction, " +
                          " pre_detail.medicine_id  medicine_id, " +
                          " pre_detail.ct  medicine_ct, " +
                          " pre.doctor_id, " +
                          " pre.patient_id, " +
                          " pre.consultation_id, " +
                          " pre.prescription_time, " +
                          " cast(pre.total_amount as decimal(16,2)) total_amount, " +
                          " cast(pre_detail.ct as bigint) * medicine.price split_amount, " +
                          " ts " +
                          "from prescription_detail pre_detail " +
                          "join  prescription pre  " +
                          "on pre_detail.prescription_id=pre.id " +
                          "join medicine for system_time as of pre_detail.pt " +
                          "on cast(pre_detail.medicine_id as bigint)=medicine.id");
        // 6.最终结果写出到 kafka
        tEnv.executeSql("create table dwd_diagnosis_prescription_detail(" +
                            " pre_detail_id string, " +
                            " prescription_id string, " +
                            " pre_detail_instruction string, " +
                            " medicine_id string, " +
                            " medicine_ct string, " +
                            " doctor_id string, " +
                            " patient_id string, " +
                            " consultation_id string, " +
                            " prescription_time string, " +
                            " total_amount decimal(16,2), " +
                            " split_amount decimal(16,2), " +
                            " ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_DIAGNOSIS_PRESCRIPTION_DETAIL));
        result.executeInsert("dwd_diagnosis_prescription_detail");
        
    }
}
